/*-------------------------------------------------------------------------
 *
 * walreceiver.c
 *
 * WAL 接收者进程（walreceiver）是 Postgres 9.0 以来的新功能。
 * 它是备用服务器中负责在流复制期间从主服务器接收
 * XLOG 记录的进程。
 *
 * 当启动进程确定是时候开始流复制时，
 * 它指示 postmaster 启动 walreceiver。Walreceiver 首先连接
 * 到主服务器（它将由主服务器中的 walsender 进程提供服务），
 * 然后只要连接存活，便持续接收 XLOG 记录并将其写入磁盘。
 * 随着 XLOG 记录的接收和刷新到磁盘，它会更新
 * WalRcv->flushedUpto 变量，以在共享内存中通知启动
 * 进程它可以在多大程度上继续进行 XLOG 重放。
 *
 * WAL 接收器无法直接加载在建立与主服务器
 * 连接时使用的 GUC 参数。相反，它依赖于在请求流时
 * 启动进程传递下来的参数值。例如，复制插槽和用于与主服务器
 * 建立连接的连接字符串。
 *
 * 如果主服务器结束流复制，但不断开连接，walreceiver
 * 进入“等待”模式，并等待启动进程提供新指令。
 * 启动进程将把其视为断开，并将重新扫描 archive/pg_wal
 * 目录。但是，当启动进程想要再次尝试流复制时，
 * 它将仅推动等待的现有 walreceiver 进程，而无需启动新的进程。
 *
 * 正常终止是通过 SIGTERM，这指示 walreceiver
 * 退出(0)。紧急终止通过 SIGQUIT；与任何 postmaster 子进程一样，
 * walreceiver 将在 SIGQUIT 时简单地中止并退出。关闭连接
 * 和 FATAL 错误被视为正常操作，而不是崩溃。
 *
 * 此文件包含 walreceiver 的面向服务器的部分。
 * libpq 特定的部分在 libpqwalreceiver 模块中。
 * 它是动态加载的，以避免将服务器与 libpq 链接。
 *
 * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/walreceiver.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <unistd.h>

#include "access/htup_details.h"
#include "access/timeline.h"
#include "access/transam.h"
#include "access/xlog_internal.h"
#include "access/xlogarchive.h"
#include "access/xlogrecovery.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_type.h"
#include "common/ip.h"
#include "funcapi.h"
#include "libpq/pqformat.h"
#include "libpq/pqsignal.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/interrupt.h"
#include "replication/walreceiver.h"
#include "replication/walsender.h"
#include "storage/ipc.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/procsignal.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/guc.h"
#include "utils/pg_lsn.h"
#include "utils/ps_status.h"
#include "utils/resowner.h"
#include "utils/timestamp.h"


/*
 * GUC 变量。 (影响 walreceiver 的其他变量在 xlog.c 中
 * 因为它们是从启动进程传递下来的，以便更好地
 * 同步。)
 */
int			wal_receiver_status_interval;
int			wal_receiver_timeout;
bool		hot_standby_feedback;

/* libpqwalreceiver 连接 */
static WalReceiverConn *wrconn = NULL;
WalReceiverFunctionsType *WalReceiverFunctions = NULL;

#define NAPTIME_PER_CYCLE 100	/* 循环之间的最大睡眠时间（100毫秒） */

/*
 * 这些变量的使用方式类似于 openLogFile/SegNo，
 * 但用于 walreceiver 写入 XLOG。 recvFileTLI 是
 * 对应于 recvFile 文件名的 TimeLineID。
 */
static int	recvFile = -1;
static TimeLineID recvFileTLI = 0;
static XLogSegNo recvSegNo = 0;

/*
 * LogstreamResult 表示我们已经
 * 写入/同步的字节位置。
 */
static struct
{
	XLogRecPtr	Write;			/* 在备用中写出的最后一个字节 + 1 */
	XLogRecPtr	Flush;			/* 在备用中刷新的最后一个字节 + 1 */
}			LogstreamResult;

static StringInfoData reply_message;
static StringInfoData incoming_message;

/* 私有函数的原型 */
static void fc_WalRcvFetchTimeLineHistoryFiles(TimeLineID fc_first, TimeLineID fc_last);
static void fc_WalRcvWaitForStartPosition(XLogRecPtr *fc_startpoint, TimeLineID *fc_startpointTLI);
static void fc_WalRcvDie(int fc_code, Datum fc_arg);
static void fc_XLogWalRcvProcessMsg(unsigned char fc_type, char *fc_buf, Size fc_len,
								 TimeLineID fc_tli);
static void fc_XLogWalRcvWrite(char *fc_buf, Size fc_nbytes, XLogRecPtr fc_recptr,
							TimeLineID fc_tli);
static void fc_XLogWalRcvFlush(bool fc_dying, TimeLineID fc_tli);
static void fc_XLogWalRcvClose(XLogRecPtr fc_recptr, TimeLineID fc_tli);
static void fc_XLogWalRcvSendReply(bool fc_force, bool fc_requestReply);
static void fc_XLogWalRcvSendHSFeedback(bool fc_immed);
static void fc_ProcessWalSndrMessage(XLogRecPtr fc_walEnd, TimestampTz fc_sendTime);

/*
 * 处理 walreceiver 进程可能收到的任何中断。
 * 只要进程的锁存器被设置，就应该调用这个函数。
 *
 * 目前，只有 SIGTERM 是值得关注的。我们不能在
 * SIGTERM 信号处理程序中直接 exit(1)，因为信号可能会在
 * 一些关键操作的中间到达，比如当我们持有自旋锁时。
 * 相反，信号处理程序会设置一个标志变量并设置
 * 进程的锁存器。每当锁存器被设置时，我们必须检查这个
 * 标志（通过调用 ProcessWalRcvInterrupts）。可能会阻塞较长
 * 时间的操作，例如从远程服务器读取，也必须关注该锁存器；
 * 请参阅 libpqrcv_PQgetResult 作为例子。
 */
void ProcessWalRcvInterrupts(void)
{
	/*
	 * 尽管 walreceiver 的中断处理与
	 * 常规后端的方案不同，但调用 CHECK_FOR_INTERRUPTS()
	 * 以确保我们在 Win32 上接收到任何传入信号，
	 * 并确保我们处理任何障碍事件。
	 */
	CHECK_FOR_INTERRUPTS();

	if (ShutdownRequestPending)
	{
		ereport(FATAL,
				(errcode(ERRCODE_ADMIN_SHUTDOWN),
				 errmsg("terminating walreceiver process due to administrator command")));
	}
}


/* walreceiver 进程的主要入口点 */
void WalReceiverMain(void)
{
	char		fc_conninfo[MAXCONNINFO];
	char	   *fc_tmp_conninfo;
	char		fc_slotname[NAMEDATALEN];
	bool		fc_is_temp_slot;
	XLogRecPtr	fc_startpoint;
	TimeLineID	fc_startpointTLI;
	TimeLineID	fc_primaryTLI;
	bool		fc_first_stream;
	WalRcvData *fc_walrcv = WalRcv;
	TimestampTz fc_last_recv_timestamp;
	TimestampTz fc_now;
	bool		fc_ping_sent;
	char	   *fc_err;
	char	   *fc_sender_host = NULL;
	int			fc_sender_port = 0;

	/*
	 * WalRcv 应该已经设置好（如果我们是后端，
	 * 我们通过 fork() 或 EXEC_BACKEND 机制从
	 * postmaster 继承这一点）。
	 */
	Assert(fc_walrcv != NULL);

	fc_now = GetCurrentTimestamp();

	/*
	 * 在共享内存中将 walreceiver 标记为正在运行。
	 *
	 * 尽早执行，这样如果我们后来失败，
	 * 我们将设置状态为 STOPPED。如果我们在此之前死掉，
	 * 启动进程将会一直等待我们启动，直到超时。
	 */
	SpinLockAcquire(&fc_walrcv->mutex);
	Assert(fc_walrcv->pid == 0);
	switch (fc_walrcv->walRcvState)
	{
		case WALRCV_STOPPING:
			/* 如果我们已经被请求停止，则不启动。 */
			fc_walrcv->walRcvState = WALRCV_STOPPED;
			/* fall through */

		case WALRCV_STOPPED:
			SpinLockRelease(&fc_walrcv->mutex);
			ConditionVariableBroadcast(&fc_walrcv->walRcvStoppedCV);
			proc_exit(1);
			break;

		case WALRCV_STARTING:
			/* 通常情况 */
			break;

		case WALRCV_WAITING:
		case WALRCV_STREAMING:
		case WALRCV_RESTARTING:
		default:
			/* 不应发生 */
			SpinLockRelease(&fc_walrcv->mutex);
			elog(PANIC, "walreceiver still running according to shared memory state");
	}
	/* 广播我们的 PID，以便启动进程可以终止我们 */
	fc_walrcv->pid = MyProcPid;
	fc_walrcv->walRcvState = WALRCV_STREAMING;

	/* 获取开始流式传输所需的信息 */
	fc_walrcv->ready_to_display = false;
	strlcpy(fc_conninfo, (char *) fc_walrcv->conninfo, MAXCONNINFO);
	strlcpy(fc_slotname, (char *) fc_walrcv->slotname, NAMEDATALEN);
	fc_is_temp_slot = fc_walrcv->is_temp_slot;
	fc_startpoint = fc_walrcv->receiveStart;
	fc_startpointTLI = fc_walrcv->receiveStartTLI;

	/*
	 * 至多只能设置 is_temp_slot 和 slotname 之一；
	 * 否则，RequestXLogStreaming 会出错。
	 */
	Assert(!fc_is_temp_slot || (fc_slotname[0] == '\0'));

	/* 初始化为合理的值 */
	fc_walrcv->lastMsgSendTime =
		fc_walrcv->lastMsgReceiptTime = fc_walrcv->latestWalEndTime = fc_now;

	/* 报告用于唤醒该进程的锁存器 */
	fc_walrcv->latch = &MyProc->procLatch;

	SpinLockRelease(&fc_walrcv->mutex);

	pg_atomic_write_u64(&WalRcv->writtenUpto, 0);

	/* 安排在 walreceiver 退出时进行清理 */
	on_shmem_exit(fc_WalRcvDie, PointerGetDatum(&fc_startpointTLI));

	/* 正确接受或忽略可能由 postmaster 发送给我们的信号 */
	pqsignal(SIGHUP, SignalHandlerForConfigReload); /* 设置标志以读取配置
													 * 文件 */
	pqsignal(SIGINT, SIG_IGN);
	pqsignal(SIGTERM, SignalHandlerForShutdownRequest); /* 请求关闭 */
	/* SIGQUIT 处理程序已经由 InitPostmasterChild 设置 */
	pqsignal(SIGALRM, SIG_IGN);
	pqsignal(SIGPIPE, SIG_IGN);
	pqsignal(SIGUSR1, procsignal_sigusr1_handler);
	pqsignal(SIGUSR2, SIG_IGN);

	/* 重置一些由 postmaster 接受但在这里不接受的信号 */
	pqsignal(SIGCHLD, SIG_DFL);

	/* 加载 libpq 特定函数 */
	load_file("libpqwalreceiver", false);
	if (WalReceiverFunctions == NULL)
		elog(ERROR, "libpqwalreceiver didn't initialize correctly");

	/* 解除信号阻塞（在 postmaster 分叉我们时信号被阻塞） */
	PG_SETMASK(&UnBlockSig);

	/* 建立与主服务器的连接以进行 XLOG 流传输 */
	wrconn = walrcv_connect(fc_conninfo, false,
							cluster_name[0] ? cluster_name : "walreceiver",
							&fc_err);
	if (!wrconn)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_FAILURE),
				 errmsg("could not connect to the primary server: %s", fc_err)));

	/*
	 * 保存用户可见的连接字符串。这会覆盖原始
	 * conninfo，出于安全考虑。还保存与这个 walreceiver 连接的
	 * 发送服务器的主机和端口。
	 */
	fc_tmp_conninfo = walrcv_get_conninfo(wrconn);
	walrcv_get_senderinfo(wrconn, &fc_sender_host, &fc_sender_port);
	SpinLockAcquire(&fc_walrcv->mutex);
	memset(fc_walrcv->conninfo, 0, MAXCONNINFO);
	if (fc_tmp_conninfo)
		strlcpy((char *) fc_walrcv->conninfo, fc_tmp_conninfo, MAXCONNINFO);

	memset(fc_walrcv->sender_host, 0, NI_MAXHOST);
	if (fc_sender_host)
		strlcpy((char *) fc_walrcv->sender_host, fc_sender_host, NI_MAXHOST);

	fc_walrcv->sender_port = fc_sender_port;
	fc_walrcv->ready_to_display = true;
	SpinLockRelease(&fc_walrcv->mutex);

	if (fc_tmp_conninfo)
		pfree(fc_tmp_conninfo);

	if (fc_sender_host)
		pfree(fc_sender_host);

	fc_first_stream = true;
	for (;;)
	{
		char	   *fc_primary_sysid;
		char		fc_standby_sysid[32];
		WalRcvStreamOptions fc_options;

		/*
		 * 检查我们是否连接到一个有效的服务器，使用
		 * IDENTIFY_SYSTEM 复制命令。
		 */
		fc_primary_sysid = walrcv_identify_system(wrconn, &fc_primaryTLI);

		snprintf(fc_standby_sysid, sizeof(fc_standby_sysid), UINT64_FORMAT,
				 GetSystemIdentifier());
		if (strcmp(fc_primary_sysid, fc_standby_sysid) != 0)
		{
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("database system identifier differs between the primary and standby"),
					 errdetail("The primary's identifier is %s, the standby's identifier is %s.",
							   fc_primary_sysid, fc_standby_sysid)));
		}

		/*
		 * 确认当前主服务器的时间线是否与我们的相同或
		 * 超前。
		 */
		if (fc_primaryTLI < fc_startpointTLI)
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("highest timeline %u of the primary is behind recovery timeline %u",
							fc_primaryTLI, fc_startpointTLI)));

		/*
		 * 获取任何缺失的历史文件。我们总是这样做，即使我们
		 * 对该时间线不感兴趣，这样如果我们后来被提升为
		 * 主服务器，我们就不会选择当前主服务器已使用的
		 * 同一时间线。这并不是万无一失的 -
		 * 如果您需要确保在每种情况下都选择唯一的时间线 ID，
		 * 您需要一些外部软件来管理您的集群，
		 * 但让我们在可以的地方避免时间线 ID 碰撞的混淆。
		 */
		fc_WalRcvFetchTimeLineHistoryFiles(fc_startpointTLI, fc_primaryTLI);

		/*
		 * 如果请求，创建临时复制槽，并在共享内存中更新槽
		 * 名称。（注意，在这种情况下，槽名称不能已经被设置。）
		 */
		if (fc_is_temp_slot)
		{
			snprintf(fc_slotname, sizeof(fc_slotname),
					 "pg_walreceiver_%lld",
					 (long long int) walrcv_get_backend_pid(wrconn));

#ifdef FDD
			walrcv_create_slot(wrconn, fc_slotname, true, false, false, 0, NULL);
#else
			walrcv_create_slot(wrconn, fc_slotname, true, false, 0, NULL);
#endif
			SpinLockAcquire(&fc_walrcv->mutex);
			strlcpy(fc_walrcv->slotname, fc_slotname, NAMEDATALEN);
			SpinLockRelease(&fc_walrcv->mutex);
		}

		/*
		 * 开始流传输。
		 *
		 * 我们会尝试在请求的起始点和时间线开始，即使它与
		 * 服务器的最新时间线不同。如果我们已经到达旧时间线的
		 * 末尾，服务器会立即完成流传输，然后我们将返回以等待
		 * 启动过程的命令。如果 recovery_target_timeline 是
		 * 'latest'，启动过程将扫描 pg_wal 并找到新的
		 * 历史文件，提升恢复目标时间线，并要求我们在新的
		 * 时间线上重新启动。
		 */
		fc_options.logical = false;
		fc_options.startpoint = fc_startpoint;
		fc_options.slotname = fc_slotname[0] != '\0' ? fc_slotname : NULL;
		fc_options.proto.physical.startpointTLI = fc_startpointTLI;
		if (walrcv_startstreaming(wrconn, &fc_options))
		{
			if (fc_first_stream)
				ereport(LOG,
						(errmsg("started streaming WAL from primary at %X/%X on timeline %u",
								LSN_FORMAT_ARGS(fc_startpoint), fc_startpointTLI)));
			else
				ereport(LOG,
						(errmsg("restarted WAL streaming at %X/%X on timeline %u",
								LSN_FORMAT_ARGS(fc_startpoint), fc_startpointTLI)));
			fc_first_stream = false;

			/* 初始化 LogstreamResult 和用于处理消息的缓冲区 */
			LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
			initStringInfo(&reply_message);
			initStringInfo(&incoming_message);

			/* 初始化最后接收的时间戳 */
			fc_last_recv_timestamp = GetCurrentTimestamp();
			fc_ping_sent = false;

			/* 循环直到流传输结束或出现错误 */
			for (;;)
			{
				char	   *fc_buf;
				int			fc_len;
				bool		fc_endofwal = false;
				pgsocket	fc_wait_fd = PGINVALID_SOCKET;
				int			fc_rc;

				/*
				 * 如果我们不在恢复中，则退出 walreceiver。这不应该
				 * 发生，但在此处进行交叉检查状态。
				 */
				if (!RecoveryInProgress())
					ereport(FATAL,
							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
							 errmsg("cannot continue WAL streaming, recovery has already ended")));

				/* 处理最近收到的任何请求或信号 */
				ProcessWalRcvInterrupts();

				if (ConfigReloadPending)
				{
					ConfigReloadPending = false;
					ProcessConfigFile(PGC_SIGHUP);
					fc_XLogWalRcvSendHSFeedback(true);
				}

				/* 查看我们是否可以立即读取数据 */
				fc_len = walrcv_receive(wrconn, &fc_buf, &fc_wait_fd);
				if (fc_len != 0)
				{
					/*
					 * 处理接收到的数据，以及我们可以在不阻塞的情况下
					 * 读取的任何后续数据。
					 */
					for (;;)
					{
						if (fc_len > 0)
						{
							/*
							 * 从主服务器收到了一些数据，因此重置
							 * 超时
							 */
							fc_last_recv_timestamp = GetCurrentTimestamp();
							fc_ping_sent = false;
							fc_XLogWalRcvProcessMsg(fc_buf[0], &fc_buf[1], fc_len - 1,
												 fc_startpointTLI);
						}
						else if (fc_len == 0)
							break;
						else if (fc_len < 0)
						{
							ereport(LOG,
									(errmsg("replication terminated by primary server"),
									 errdetail("End of WAL reached on timeline %u at %X/%X.",
											   fc_startpointTLI,
											   LSN_FORMAT_ARGS(LogstreamResult.Write))));
							fc_endofwal = true;
							break;
						}
						fc_len = walrcv_receive(wrconn, &fc_buf, &fc_wait_fd);
					}

					/* 让主服务器知道我们收到了数据。 */
					fc_XLogWalRcvSendReply(false, false);

					/*
					 * 如果我们写了一些记录，将它们刷新到磁盘，并
					 * 让启动过程和主服务器知道它们。
					 */
					fc_XLogWalRcvFlush(false, fc_startpointTLI);
				}

				/* 检查我们是否需要退出流传输循环。 */
				if (fc_endofwal)
					break;

				/*
				 * 理想情况下，我们希望在这里重复使用一个 WaitEventSet 对象
				 * 来避免在 epoll 系统上 WaitLatchOrSocket 的开销，但我们不能确保 libpq
				 * （或其他任何 walreceiver 实现）具有相同的套接字（即使 fd 是相同的数字，
				 * 它在上次使用后可能已经关闭并重新打开）。将来，如果有一个
				 * 移除套接字的 WaitEventSet 函数，我们就可以每次只添加和移除套接字，
				 * 从而可能避免一些系统调用。
				 */
				Assert(fc_wait_fd != PGINVALID_SOCKET);
				fc_rc = WaitLatchOrSocket(MyLatch,
									   WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
									   WL_TIMEOUT | WL_LATCH_SET,
									   fc_wait_fd,
									   NAPTIME_PER_CYCLE,
									   WAIT_EVENT_WAL_RECEIVER_MAIN);
				if (fc_rc & WL_LATCH_SET)
				{
					ResetLatch(MyLatch);
					ProcessWalRcvInterrupts();

					if (fc_walrcv->force_reply)
					{
						/*
						 * 恢复过程请求我们现在发送应用反馈。确保在发送回复之前，
						 * 共享内存中的标志确实被设置为 false，以便我们不会错过新的
						 * 回复请求。
						 */
						fc_walrcv->force_reply = false;
						pg_memory_barrier();
						fc_XLogWalRcvSendReply(true, false);
					}
				}
				if (fc_rc & WL_TIMEOUT)
				{
					/*
					 * 我们没有收到任何新消息。如果我们已经超过
					 * wal_receiver_timeout / 2 没有收到来自服务器的消息，
					 * 则 ping 服务器。此外，如果自上次我们发送的更新以来，
					 * 超过 wal_receiver_status_interval 的时间，仍然发送状态更新
					 * 给主节点，以报告应用 WAL 的任何进展。
					 */
					bool		fc_requestReply = false;

					/*
					 * 检查自上次从主节点接收数据以来的时间是否已达到
					 * 配置的限制。
					 */
					if (wal_receiver_timeout > 0)
					{
						TimestampTz fc_now = GetCurrentTimestamp();
						TimestampTz fc_timeout;

						fc_timeout =
							TimestampTzPlusMilliseconds(fc_last_recv_timestamp,
														wal_receiver_timeout);

						if (fc_now >= fc_timeout)
							ereport(ERROR,
									(errcode(ERRCODE_CONNECTION_FAILURE),
									 errmsg("terminating walreceiver due to timeout")));

						/*
						 * 我们没有收到任何新消息，超过接收器复制超时的一半。
						 * Ping 服务器。
						 */
						if (!fc_ping_sent)
						{
							fc_timeout = TimestampTzPlusMilliseconds(fc_last_recv_timestamp,
																  (wal_receiver_timeout / 2));
							if (fc_now >= fc_timeout)
							{
								fc_requestReply = true;
								fc_ping_sent = true;
							}
						}
					}

					fc_XLogWalRcvSendReply(fc_requestReply, fc_requestReply);
					fc_XLogWalRcvSendHSFeedback(false);
				}
			}

			/*
			 * 后端完成了流式传输。从我们这边也退出流式 COPY 模式。
			 */
			walrcv_endstreaming(wrconn, &fc_primaryTLI);

			/*
			 * 如果服务器在我们开始流式传输时切换到了一个我们不知道的新时间线，
			 * 现在获取它的时间线历史文件。
			 */
			fc_WalRcvFetchTimeLineHistoryFiles(fc_startpointTLI, fc_primaryTLI);
		}
		else
			ereport(LOG,
					(errmsg("primary server contains no more WAL on requested timeline %u",
							fc_startpointTLI)));

		/*
		 * 在请求的时间线中到达 WAL 的末尾。关闭最后一个段，并
		 * 等待启动进程的新命令。
		 */
		if (recvFile >= 0)
		{
			char		fc_xlogfname[MAXFNAMELEN];

			fc_XLogWalRcvFlush(false, fc_startpointTLI);
			XLogFileName(fc_xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
			if (close(recvFile) != 0)
				ereport(PANIC,
						(errcode_for_file_access(),
						 errmsg("could not close log segment %s: %m",
								fc_xlogfname)));

			/*
			 * 强制创建 .done 文件，以防止流式段被后续归档。
			 */
			if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
				XLogArchiveForceDone(fc_xlogfname);
			else
				XLogArchiveNotify(fc_xlogfname);
		}
		recvFile = -1;

		elog(DEBUG1, "walreceiver ended streaming and awaits new instructions");
		fc_WalRcvWaitForStartPosition(&fc_startpoint, &fc_startpointTLI);
	}
	/* 未到达 */
}

/*
 * 等待启动进程设置 receiveStart 和 receiveStartTLI。
 */
static void fc_WalRcvWaitForStartPosition(XLogRecPtr *fc_startpoint, TimeLineID *fc_startpointTLI)
{
	WalRcvData *fc_walrcv = WalRcv;
	int			fc_state;

	SpinLockAcquire(&fc_walrcv->mutex);
	fc_state = fc_walrcv->walRcvState;
	if (fc_state != WALRCV_STREAMING)
	{
		SpinLockRelease(&fc_walrcv->mutex);
		if (fc_state == WALRCV_STOPPING)
			proc_exit(0);
		else
			elog(FATAL, "unexpected walreceiver state");
	}
	fc_walrcv->walRcvState = WALRCV_WAITING;
	fc_walrcv->receiveStart = InvalidXLogRecPtr;
	fc_walrcv->receiveStartTLI = 0;
	SpinLockRelease(&fc_walrcv->mutex);

	set_ps_display("idle");

	/*
	 * 提醒启动进程注意我们已经停止流式传输并在等待指令。
	 */
	WakeupRecovery();
	for (;;)
	{
		ResetLatch(MyLatch);

		ProcessWalRcvInterrupts();

		SpinLockAcquire(&fc_walrcv->mutex);
		Assert(fc_walrcv->walRcvState == WALRCV_RESTARTING ||
			   fc_walrcv->walRcvState == WALRCV_WAITING ||
			   fc_walrcv->walRcvState == WALRCV_STOPPING);
		if (fc_walrcv->walRcvState == WALRCV_RESTARTING)
		{
			/*
			 * 在这里不需要处理 primary_conninfo 或 primary_slot_name 的变化。
			 * 启动进程将在这些变化时通知我们终止。
			 */
			*fc_startpoint = fc_walrcv->receiveStart;
			*fc_startpointTLI = fc_walrcv->receiveStartTLI;
			fc_walrcv->walRcvState = WALRCV_STREAMING;
			SpinLockRelease(&fc_walrcv->mutex);
			break;
		}
		if (fc_walrcv->walRcvState == WALRCV_STOPPING)
		{
			/*
			 * 如果启动进程希望我们停止，应该已经收到 SIGTERM，
			 * 但在这里检查一下也是可以的。
			 */
			SpinLockRelease(&fc_walrcv->mutex);
			exit(1);
		}
		SpinLockRelease(&fc_walrcv->mutex);

		(void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
						 WAIT_EVENT_WAL_RECEIVER_WAIT_START);
	}

	if (update_process_title)
	{
		char		fc_activitymsg[50];

		snprintf(fc_activitymsg, sizeof(fc_activitymsg), "restarting at %X/%X",
				 LSN_FORMAT_ARGS(*fc_startpoint));
		set_ps_display(fc_activitymsg);
	}
}

/*
 * 从服务器获取 'first' 和 'last' （包含）之间的任何缺失的时间线历史文件。
 */
static void fc_WalRcvFetchTimeLineHistoryFiles(TimeLineID fc_first, TimeLineID fc_last)
{
	TimeLineID	fc_tli;

	for (fc_tli = fc_first; fc_tli <= fc_last; fc_tli++)
	{
		/* 时间线 1 没有历史文件 */
		if (fc_tli != 1 && !existsTimeLineHistory(fc_tli))
		{
			char	   *fc_fname;
			char	   *fc_content;
			int			fc_len;
			char		fc_expectedfname[MAXFNAMELEN];

			ereport(LOG,
					(errmsg("fetching timeline history file for timeline %u from primary server",
							fc_tli)));

			walrcv_readtimelinehistoryfile(wrconn, fc_tli, &fc_fname, &fc_content, &fc_len);

			/*
			 * 检查主节点上的文件名是否与我们自己计算的匹配。
			 * 这只是一个合理性检查，应该始终匹配。
			 */
			TLHistoryFileName(fc_expectedfname, fc_tli);
			if (strcmp(fc_fname, fc_expectedfname) != 0)
				ereport(ERROR,
						(errcode(ERRCODE_PROTOCOL_VIOLATION),
						 errmsg_internal("primary reported unexpected file name for timeline history file of timeline %u",
										 fc_tli)));

			/*
			 * 将文件写入 pg_wal。
			 */
			writeTimeLineHistoryFile(fc_tli, fc_content, fc_len);

			/*
			 * 如果 archive_mode 为 always，则将流式历史文件标记为
			 * 准备归档。
			 */
			if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
				XLogArchiveForceDone(fc_fname);
			else
				XLogArchiveNotify(fc_fname);

			pfree(fc_fname);
			pfree(fc_content);
		}
	}
}

/*
 * 在退出时将我们标记为在共享内存中停止。
 */
static void fc_WalRcvDie(int fc_code, Datum fc_arg)
{
	WalRcvData *fc_walrcv = WalRcv;
	TimeLineID *fc_startpointTLI_p = (TimeLineID *) DatumGetPointer(fc_arg);

	Assert(*fc_startpointTLI_p != 0);

	/* 确保所有接收到的WAL记录已刷新到磁盘 */
	fc_XLogWalRcvFlush(true, *fc_startpointTLI_p);

	/* 在共享内存中标记自己为非活动状态 */
	SpinLockAcquire(&fc_walrcv->mutex);
	Assert(fc_walrcv->walRcvState == WALRCV_STREAMING ||
		   fc_walrcv->walRcvState == WALRCV_RESTARTING ||
		   fc_walrcv->walRcvState == WALRCV_STARTING ||
		   fc_walrcv->walRcvState == WALRCV_WAITING ||
		   fc_walrcv->walRcvState == WALRCV_STOPPING);
	Assert(fc_walrcv->pid == MyProcPid);
	fc_walrcv->walRcvState = WALRCV_STOPPED;
	fc_walrcv->pid = 0;
	fc_walrcv->ready_to_display = false;
	fc_walrcv->latch = NULL;
	SpinLockRelease(&fc_walrcv->mutex);

	ConditionVariableBroadcast(&fc_walrcv->walRcvStoppedCV);

	/* 优雅地终止连接。 */
	if (wrconn != NULL)
		walrcv_disconnect(wrconn);

	/* 唤醒启动进程以迅速注意到我们已经离开 */
	WakeupRecovery();
}

/*
 * 接受来自XLOG流的消息，并处理它。
 */
static void fc_XLogWalRcvProcessMsg(unsigned char fc_type, char *fc_buf, Size fc_len, TimeLineID fc_tli)
{
	int			fc_hdrlen;
	XLogRecPtr	fc_dataStart;
	XLogRecPtr	fc_walEnd;
	TimestampTz fc_sendTime;
	bool		fc_replyRequested;

	resetStringInfo(&incoming_message);

	switch (fc_type)
	{
		case 'w':				/* WAL记录 */
			{
				/* 将消息复制到StringInfo */
				fc_hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64);
				if (fc_len < fc_hdrlen)
					ereport(ERROR,
							(errcode(ERRCODE_PROTOCOL_VIOLATION),
							 errmsg_internal("invalid WAL message received from primary")));
				appendBinaryStringInfo(&incoming_message, fc_buf, fc_hdrlen);

				/* 读取字段 */
				fc_dataStart = pq_getmsgint64(&incoming_message);
				fc_walEnd = pq_getmsgint64(&incoming_message);
				fc_sendTime = pq_getmsgint64(&incoming_message);
				fc_ProcessWalSndrMessage(fc_walEnd, fc_sendTime);

				fc_buf += fc_hdrlen;
				fc_len -= fc_hdrlen;
				fc_XLogWalRcvWrite(fc_buf, fc_len, fc_dataStart, fc_tli);
				break;
			}
		case 'k':				/* 保持连接 */
			{
				/* 将消息复制到StringInfo */
				fc_hdrlen = sizeof(int64) + sizeof(int64) + sizeof(char);
				if (fc_len != fc_hdrlen)
					ereport(ERROR,
							(errcode(ERRCODE_PROTOCOL_VIOLATION),
							 errmsg_internal("invalid keepalive message received from primary")));
				appendBinaryStringInfo(&incoming_message, fc_buf, fc_hdrlen);

				/* 读取字段 */
				fc_walEnd = pq_getmsgint64(&incoming_message);
				fc_sendTime = pq_getmsgint64(&incoming_message);
				fc_replyRequested = pq_getmsgbyte(&incoming_message);

				fc_ProcessWalSndrMessage(fc_walEnd, fc_sendTime);

				/* 如果主服务器请求回复，立即发送一条 */
				if (fc_replyRequested)
					fc_XLogWalRcvSendReply(true, false);
				break;
			}
		default:
			ereport(ERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg_internal("invalid replication message type %d",
									 fc_type)));
	}
}

/*
 * 将XLOG数据写入磁盘。
 */
static void fc_XLogWalRcvWrite(char *fc_buf, Size fc_nbytes, XLogRecPtr fc_recptr, TimeLineID fc_tli)
{
	int			fc_startoff;
	int			fc_byteswritten;

	Assert(fc_tli != 0);

	while (fc_nbytes > 0)
	{
		int			fc_segbytes;

		/* 如果当前段完成，则关闭当前段 */
		if (recvFile >= 0 && !XLByteInSeg(fc_recptr, recvSegNo, wal_segment_size))
			fc_XLogWalRcvClose(fc_recptr, fc_tli);

		if (recvFile < 0)
		{
			/* 创建/使用新的日志文件 */
			XLByteToSeg(fc_recptr, recvSegNo, wal_segment_size);
			recvFile = XLogFileInit(recvSegNo, fc_tli);
			recvFileTLI = fc_tli;
		}

		/* 计算接收到的日志的起始偏移量 */
		fc_startoff = XLogSegmentOffset(fc_recptr, wal_segment_size);

		if (fc_startoff + fc_nbytes > wal_segment_size)
			fc_segbytes = wal_segment_size - fc_startoff;
		else
			fc_segbytes = fc_nbytes;

		/* 可以写入日志 */
		errno = 0;

		fc_byteswritten = pg_pwrite(recvFile, fc_buf, fc_segbytes, (off_t) fc_startoff);
		if (fc_byteswritten <= 0)
		{
			char		fc_xlogfname[MAXFNAMELEN];
			int			fc_save_errno;

			/* 如果写入没有设置 errno，假设没有磁盘空间 */
			if (errno == 0)
				errno = ENOSPC;

			fc_save_errno = errno;
			XLogFileName(fc_xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
			errno = fc_save_errno;
			ereport(PANIC,
					(errcode_for_file_access(),
					 errmsg("could not write to log segment %s "
							"at offset %u, length %lu: %m",
							fc_xlogfname, fc_startoff, (unsigned long) fc_segbytes)));
		}

		/* 更新写入的状态 */
		fc_recptr += fc_byteswritten;

		fc_nbytes -= fc_byteswritten;
		fc_buf += fc_byteswritten;

		LogstreamResult.Write = fc_recptr;
	}

	/* 更新共享内存状态 */
	pg_atomic_write_u64(&WalRcv->writtenUpto, LogstreamResult.Write);

	/*
	 * 如果在最后一个循环周期中当前段已完全写入，则关闭当前段，
	 * 以便尽快创建其归档通知文件。否则，当前段的WAL归档将被延迟，
	 * 直到接收到并写入下一个段中的任何数据。
	 */
	if (recvFile >= 0 && !XLByteInSeg(fc_recptr, recvSegNo, wal_segment_size))
		fc_XLogWalRcvClose(fc_recptr, fc_tli);
}

/*
 * 将日志刷新到磁盘。
 *
 * 如果我们正在死亡之中，做任何可能引发错误的事情都是不明智的，
 * 所以在这种情况下我们跳过发送回复。
 */
static void fc_XLogWalRcvFlush(bool fc_dying, TimeLineID fc_tli)
{
	Assert(fc_tli != 0);

	if (LogstreamResult.Flush < LogstreamResult.Write)
	{
		WalRcvData *fc_walrcv = WalRcv;

		issue_xlog_fsync(recvFile, recvSegNo, fc_tli);

		LogstreamResult.Flush = LogstreamResult.Write;

		/* 更新共享内存状态 */
		SpinLockAcquire(&fc_walrcv->mutex);
		if (fc_walrcv->flushedUpto < LogstreamResult.Flush)
		{
			fc_walrcv->latestChunkStart = fc_walrcv->flushedUpto;
			fc_walrcv->flushedUpto = LogstreamResult.Flush;
			fc_walrcv->receivedTLI = fc_tli;
		}
		SpinLockRelease(&fc_walrcv->mutex);

		/* 通知启动进程和walsender新的WAL已到达 */
		WakeupRecovery();
		if (AllowCascadeReplication())
			WalSndWakeup();

		/* 在PS显示中报告XLOG流进度 */
		if (update_process_title)
		{
			char		fc_activitymsg[50];

			snprintf(fc_activitymsg, sizeof(fc_activitymsg), "streaming %X/%X",
					 LSN_FORMAT_ARGS(LogstreamResult.Write));
			set_ps_display(fc_activitymsg);
		}

		/* 还让主服务器知道我们取得了一些进展 */
		if (!fc_dying)
		{
			fc_XLogWalRcvSendReply(false, false);
			fc_XLogWalRcvSendHSFeedback(false);
		}
	}
}

/*
 * 关闭当前段。
 *
 * 在关闭它之前将段刷新到磁盘。否则我们必须在稍后重新打开并fsync它。
 *
 * 创建一个归档通知文件，因为段已知为完成。
 */
static void fc_XLogWalRcvClose(XLogRecPtr fc_recptr, TimeLineID fc_tli)
{
	char		fc_xlogfname[MAXFNAMELEN];

	Assert(recvFile >= 0 && !XLByteInSeg(fc_recptr, recvSegNo, wal_segment_size));
	Assert(fc_tli != 0);

	/*
	 * 在切换到下一个文件之前fsync()并关闭当前文件。否则我们
	 * 必须稍后重新打开此文件以进行fsync。
	 */
	fc_XLogWalRcvFlush(false, fc_tli);

	XLogFileName(fc_xlogfname, recvFileTLI, recvSegNo, wal_segment_size);

	/*
	 * XLOG段文件将在启动进程中被恢复重新读取，
	 * 所以我们不建议操作系统释放与该文件相关的缓存页，
	 * 就像XLogFileClose()所做的那样。
	 */
	if (close(recvFile) != 0)
		ereport(PANIC,
				(errcode_for_file_access(),
				 errmsg("could not close log segment %s: %m",
						fc_xlogfname)));

	/*
	 * 强制创建.done文件，以防止流式段稍后被归档。
	 */
	if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
		XLogArchiveForceDone(fc_xlogfname);
	else
		XLogArchiveNotify(fc_xlogfname);

	recvFile = -1;
}

/*
 * 向主服务器发送回复消息，指示我们当前的WAL位置，
 * 最旧的xmin和当前时间。
 *
 * 如果未设置'force'，则仅在自上次状态更新以来经过足够时间的情况下发送消息，
 * 以达到wal_receiver_status_interval。
 * 如果完全禁用wal_receiver_status_interval并且'force'为false，
 * 则这是一个无操作。
 *
 * 如果'requestReply'为true，请求服务器在收到此消息后立即回复。
 * 这用于心跳，当接近wal_receiver_timeout时。
 */
static void fc_XLogWalRcvSendReply(bool fc_force, bool fc_requestReply)
{
	static XLogRecPtr fc_writePtr = 0;
	static XLogRecPtr fc_flushPtr = 0;
	XLogRecPtr	fc_applyPtr;
	static TimestampTz fc_sendTime = 0;
	TimestampTz fc_now;

	/*
	 * 如果用户不希望状态报告给主服务器，请确保
	 * 在做任何事情之前退出。
	 */
	if (!fc_force && wal_receiver_status_interval <= 0)
		return;

	/* 获取当前时间戳。 */
	fc_now = GetCurrentTimestamp();

	/*
	 * 我们可以在不获取任何锁的情况下比较写入和刷新位置与
	 * 我们发送的最后一条消息，但应用位置需要一个自旋锁，
	 * 所以除非其他情况发生变化或已经过去10秒，否则我们不检查它。
	 * 这意味着从主节点的角度来看，应用WAL位置看起来会稍微延迟，
	 * 但由于这仅用于报告目的，并且仅在空闲系统上，这可能是可以的。
	 */
	if (!fc_force
		&& fc_writePtr == LogstreamResult.Write
		&& fc_flushPtr == LogstreamResult.Flush
		&& !TimestampDifferenceExceeds(fc_sendTime, fc_now,
									   wal_receiver_status_interval * 1000))
		return;
	fc_sendTime = fc_now;

	/* 构造一条新消息 */
	fc_writePtr = LogstreamResult.Write;
	fc_flushPtr = LogstreamResult.Flush;
	fc_applyPtr = GetXLogReplayRecPtr(NULL);

	resetStringInfo(&reply_message);
	pq_sendbyte(&reply_message, 'r');
	pq_sendint64(&reply_message, fc_writePtr);
	pq_sendint64(&reply_message, fc_flushPtr);
	pq_sendint64(&reply_message, fc_applyPtr);
	pq_sendint64(&reply_message, GetCurrentTimestamp());
	pq_sendbyte(&reply_message, fc_requestReply ? 1 : 0);

	/* 发送它 */
	elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X%s",
		 LSN_FORMAT_ARGS(fc_writePtr),
		 LSN_FORMAT_ARGS(fc_flushPtr),
		 LSN_FORMAT_ARGS(fc_applyPtr),
		 fc_requestReply ? " (reply requested)" : "");

	walrcv_send(wrconn, reply_message.data, reply_message.len);
}

/*
 * 发送热备用反馈消息到主节点，以及当前时间，
 * 以防他们没有手表。
 *
 * 如果用户禁用反馈，发送一条最终消息告诉发送者
 * 忘记这个备用的xmin。我们在首次连接时也发送此消息，
 * 因为之前的连接可能已经在复制槽上设置了xmin。
 * （如果我们没有使用槽，显式发送一个设置InvalidTransactionId的反馈消息是无害的）。
 */
static void fc_XLogWalRcvSendHSFeedback(bool fc_immed)
{
	TimestampTz fc_now;
	FullTransactionId fc_nextFullXid;
	TransactionId fc_nextXid;
	uint32		fc_xmin_epoch,
				fc_catalog_xmin_epoch;
	TransactionId fc_xmin,
				fc_catalog_xmin;
	static TimestampTz fc_sendTime = 0;

	/* 最初为true，因此我们始终发送至少一条反馈消息 */
	static bool fc_primary_has_standby_xmin = true;

	/*
	 * 如果用户不希望状态报告给主服务器，请确保
	 * 在做任何事情之前退出。
	 */
	if ((wal_receiver_status_interval <= 0 || !hot_standby_feedback) &&
		!fc_primary_has_standby_xmin)
		return;

	/* 获取当前时间戳。 */
	fc_now = GetCurrentTimestamp();

	if (!fc_immed)
	{
		/*
		 * 每个wal_receiver_status_interval最多发送一次反馈。
		 */
		if (!TimestampDifferenceExceeds(fc_sendTime, fc_now,
										wal_receiver_status_interval * 1000))
			return;
		fc_sendTime = fc_now;
	}

	/*
	 * 如果热备用还没有接受连接，则没有任何东西可发送。
	 * 在间隔到期后检查这一点以减少调用次数。
	 *
	 * 在这里退出还确保我们在读取自己的复制槽状态之前
	 * 不发送反馈，因此我们不会告诉主节点丢弃任何可能存在于
	 * 此副本上的槽中所需的xmin或catalog_xmin。
	 */
	if (!HotStandbyActive())
		return;

	/*
	 * 一旦我们确定其他所有检查都已完成，进行获取最旧xmin的高成本调用。
	 */
	if (hot_standby_feedback)
	{
		GetReplicationHorizons(&fc_xmin, &fc_catalog_xmin);
	}
	else
	{
		fc_xmin = InvalidTransactionId;
		fc_catalog_xmin = InvalidTransactionId;
	}

	/*
	 * 获取纪元，并在nextXid和oldestXmin位于纪元边界的不同两侧时进行调整。
	 */
	fc_nextFullXid = ReadNextFullTransactionId();
	fc_nextXid = XidFromFullTransactionId(fc_nextFullXid);
	fc_xmin_epoch = EpochFromFullTransactionId(fc_nextFullXid);
	fc_catalog_xmin_epoch = fc_xmin_epoch;
	if (fc_nextXid < fc_xmin)
		fc_xmin_epoch--;
	if (fc_nextXid < fc_catalog_xmin)
		fc_catalog_xmin_epoch--;

	elog(DEBUG2, "sending hot standby feedback xmin %u epoch %u catalog_xmin %u catalog_xmin_epoch %u",
		 fc_xmin, fc_xmin_epoch, fc_catalog_xmin, fc_catalog_xmin_epoch);

	/* 构造消息并发送。 */
	resetStringInfo(&reply_message);
	pq_sendbyte(&reply_message, 'h');
	pq_sendint64(&reply_message, GetCurrentTimestamp());
	pq_sendint32(&reply_message, fc_xmin);
	pq_sendint32(&reply_message, fc_xmin_epoch);
	pq_sendint32(&reply_message, fc_catalog_xmin);
	pq_sendint32(&reply_message, fc_catalog_xmin_epoch);
	walrcv_send(wrconn, reply_message.data, reply_message.len);
	if (TransactionIdIsValid(fc_xmin) || TransactionIdIsValid(fc_catalog_xmin))
		fc_primary_has_standby_xmin = true;
	else
		fc_primary_has_standby_xmin = false;
}

/*
 * 在接收到来自主节点的消息后更新共享内存状态。
 *
 * 'walEnd'和'sendTime'是主节点报告的WAL结束和最新消息的时间戳。
 */
static void fc_ProcessWalSndrMessage(XLogRecPtr fc_walEnd, TimestampTz fc_sendTime)
{
	WalRcvData *fc_walrcv = WalRcv;

	TimestampTz fc_lastMsgReceiptTime = GetCurrentTimestamp();

	/* 更新共享内存状态 */
	SpinLockAcquire(&fc_walrcv->mutex);
	if (fc_walrcv->latestWalEnd < fc_walEnd)
		fc_walrcv->latestWalEndTime = fc_sendTime;
	fc_walrcv->latestWalEnd = fc_walEnd;
	fc_walrcv->lastMsgSendTime = fc_sendTime;
	fc_walrcv->lastMsgReceiptTime = fc_lastMsgReceiptTime;
	SpinLockRelease(&fc_walrcv->mutex);

	if (message_level_is_interesting(DEBUG2))
	{
		char	   *fc_sendtime;
		char	   *fc_receipttime;
		int			fc_applyDelay;

		/* 复制因为timestamptz_to_str返回一个静态缓冲区 */
		fc_sendtime = pstrdup(timestamptz_to_str(fc_sendTime));
		fc_receipttime = pstrdup(timestamptz_to_str(fc_lastMsgReceiptTime));
		fc_applyDelay = GetReplicationApplyDelay();

		/* apply延迟不可用 */
		if (fc_applyDelay == -1)
			elog(DEBUG2, "sendtime %s receipttime %s replication apply delay (N/A) transfer latency %d ms",
				 fc_sendtime,
				 fc_receipttime,
				 GetReplicationTransferLatency());
		else
			elog(DEBUG2, "sendtime %s receipttime %s replication apply delay %d ms transfer latency %d ms",
				 fc_sendtime,
				 fc_receipttime,
				 fc_applyDelay,
				 GetReplicationTransferLatency());

		pfree(fc_sendtime);
		pfree(fc_receipttime);
	}
}

/*
 * 唤醒walreceiver主循环。
 *
 * 每当应用有趣的xlog记录时，启动进程会调用此函数，
 * 以便walreceiver可以检查是否需要向主节点发送
 * 应用通知，主节点可能正在等待与
 * synchronous_commit = remote_apply的COMMIT。
 */
void WalRcvForceReply(void)
{
	Latch	   *fc_latch;

	WalRcv->force_reply = true;
	/* 获取闩锁指针可能不是原子的，所以使用自旋锁 */
	SpinLockAcquire(&WalRcv->mutex);
	fc_latch = WalRcv->latch;
	SpinLockRelease(&WalRcv->mutex);
	if (fc_latch)
		SetLatch(fc_latch);
}

/*
 * 返回表示状态的字符串常量。此字符串在系统函数和视图中使用，
 * 且*不*应翻译。
 */
static const char * fc_WalRcvGetStateString(WalRcvState fc_state)
{
	switch (fc_state)
	{
		case WALRCV_STOPPED:
			return "stopped";
		case WALRCV_STARTING:
			return "starting";
		case WALRCV_STREAMING:
			return "streaming";
		case WALRCV_WAITING:
			return "waiting";
		case WALRCV_RESTARTING:
			return "restarting";
		case WALRCV_STOPPING:
			return "stopping";
	}
	return "UNKNOWN";
}

/*
 * 返回WAL接收器的活动，包括从另一个服务器的WAL发送器接收到的pid、状态和xlog位置。
 */
Datum pg_stat_get_wal_receiver(PG_FUNCTION_ARGS)
{
	TupleDesc	fc_tupdesc;
	Datum	   *fc_values;
	bool	   *fc_nulls;
	int			fc_pid;
	bool		fc_ready_to_display;
	WalRcvState fc_state;
	XLogRecPtr	fc_receive_start_lsn;
	TimeLineID	fc_receive_start_tli;
	XLogRecPtr	fc_written_lsn;
	XLogRecPtr	fc_flushed_lsn;
	TimeLineID	fc_received_tli;
	TimestampTz fc_last_send_time;
	TimestampTz fc_last_receipt_time;
	XLogRecPtr	fc_latest_end_lsn;
	TimestampTz fc_latest_end_time;
	char		fc_sender_host[NI_MAXHOST];
	int			fc_sender_port = 0;
	char		fc_slotname[NAMEDATALEN];
	char		fc_conninfo[MAXCONNINFO];

	/* 获取锁以确保值一致性 */
	SpinLockAcquire(&WalRcv->mutex);
	fc_pid = (int) WalRcv->pid;
	fc_ready_to_display = WalRcv->ready_to_display;
	fc_state = WalRcv->walRcvState;
	fc_receive_start_lsn = WalRcv->receiveStart;
	fc_receive_start_tli = WalRcv->receiveStartTLI;
	fc_flushed_lsn = WalRcv->flushedUpto;
	fc_received_tli = WalRcv->receivedTLI;
	fc_last_send_time = WalRcv->lastMsgSendTime;
	fc_last_receipt_time = WalRcv->lastMsgReceiptTime;
	fc_latest_end_lsn = WalRcv->latestWalEnd;
	fc_latest_end_time = WalRcv->latestWalEndTime;
	strlcpy(fc_slotname, (char *) WalRcv->slotname, sizeof(fc_slotname));
	strlcpy(fc_sender_host, (char *) WalRcv->sender_host, sizeof(fc_sender_host));
	fc_sender_port = WalRcv->sender_port;
	strlcpy(fc_conninfo, (char *) WalRcv->conninfo, sizeof(fc_conninfo));
	SpinLockRelease(&WalRcv->mutex);

	/*
	 * 没有WAL接收器（或尚未准备好），只返回一个包含NULL值的元组。
	 */
	if (fc_pid == 0 || !fc_ready_to_display)
		PG_RETURN_NULL();

	/*
	 * 在不持有自旋锁的情况下读取"writtenUpto"。请注意，它可能与WAL接收器的其他受自旋锁保护的共享变量不一致，但这不应用于数据完整性检查。
	 */
	fc_written_lsn = pg_atomic_read_u64(&WalRcv->writtenUpto);

	/* 确定结果类型 */
	if (get_call_result_type(fcinfo, NULL, &fc_tupdesc) != TYPEFUNC_COMPOSITE)
		elog(ERROR, "return type must be a row type");

	fc_values = palloc0(sizeof(Datum) * fc_tupdesc->natts);
	fc_nulls = palloc0(sizeof(bool) * fc_tupdesc->natts);

	/* 获取值 */
	fc_values[0] = Int32GetDatum(fc_pid);

	if (!has_privs_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS))
	{
		/*
		 * 只有超级用户和具有pg_read_all_stats权限的角色才能看到详细信息。其他用户只获取pid值以了解它是否是WAL接收器，但没有详细信息。
		 */
		MemSet(&fc_nulls[1], true, sizeof(bool) * (fc_tupdesc->natts - 1));
	}
	else
	{
		fc_values[1] = CStringGetTextDatum(fc_WalRcvGetStateString(fc_state));

		if (XLogRecPtrIsInvalid(fc_receive_start_lsn))
			fc_nulls[2] = true;
		else
			fc_values[2] = LSNGetDatum(fc_receive_start_lsn);
		fc_values[3] = Int32GetDatum(fc_receive_start_tli);
		if (XLogRecPtrIsInvalid(fc_written_lsn))
			fc_nulls[4] = true;
		else
			fc_values[4] = LSNGetDatum(fc_written_lsn);
		if (XLogRecPtrIsInvalid(fc_flushed_lsn))
			fc_nulls[5] = true;
		else
			fc_values[5] = LSNGetDatum(fc_flushed_lsn);
		fc_values[6] = Int32GetDatum(fc_received_tli);
		if (fc_last_send_time == 0)
			fc_nulls[7] = true;
		else
			fc_values[7] = TimestampTzGetDatum(fc_last_send_time);
		if (fc_last_receipt_time == 0)
			fc_nulls[8] = true;
		else
			fc_values[8] = TimestampTzGetDatum(fc_last_receipt_time);
		if (XLogRecPtrIsInvalid(fc_latest_end_lsn))
			fc_nulls[9] = true;
		else
			fc_values[9] = LSNGetDatum(fc_latest_end_lsn);
		if (fc_latest_end_time == 0)
			fc_nulls[10] = true;
		else
			fc_values[10] = TimestampTzGetDatum(fc_latest_end_time);
		if (*fc_slotname == '\0')
			fc_nulls[11] = true;
		else
			fc_values[11] = CStringGetTextDatum(fc_slotname);
		if (*fc_sender_host == '\0')
			fc_nulls[12] = true;
		else
			fc_values[12] = CStringGetTextDatum(fc_sender_host);
		if (fc_sender_port == 0)
			fc_nulls[13] = true;
		else
			fc_values[13] = Int32GetDatum(fc_sender_port);
		if (*fc_conninfo == '\0')
			fc_nulls[14] = true;
		else
			fc_values[14] = CStringGetTextDatum(fc_conninfo);
	}

	/* 将记录作为 Datum 返回 */
	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(fc_tupdesc, fc_values, fc_nulls)));
}
